package com.baishancloud.log.douyin

import com.baishancloud.log.common.util.other.DataTimeFilterUtil
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration

import java.util.{Objects, Timer, TimerTask}

/**
 * 延迟数据过滤、以及需求中的过滤，需求1和需求2的过滤条件完全一致<br>
 * 由于头条  域名数据从金华地区切换到石家庄地区 kafka 集群，因此抖音的请求需要两个地方进行写入，故在原来的基础上，在 tag 标签中增加 kafka_source 字段来群是消费的哪个地区的 kafka 集群数据。<br>
 * 该转换，对运行在石家庄地区的任务数据进行转换，其中包含了上述域名的过滤逻辑。<br>
 * 过滤：指定消费该域名的开始时间
 *
 * @author ziqiang.wang 
 * @date 2022/3/1 14:25
 */
class DataFilterSjz(parameterTool: ParameterTool) extends RichFilterFunction[Origin] {

  var allowLatest: Long = System.currentTimeMillis() - 24 * 3600 * 1000L

  var filterDomainsSet: Set[String] = _
  var minSecond: Long = Long.MinValue

  override def open(parameters: Configuration): Unit = {
    // 按照最小时间顾虑指定域名数据
    filterDomainsSet = parameterTool.get(filterDomains).split(",").toSet
    minSecond = parameterTool.getLong(filterSecond)
    new Timer("更新当前时间", true).schedule(new TimerTask {
      override def run(): Unit = allowLatest = System.currentTimeMillis() - 24 * 3600 * 1000L
    }, 60 * 1000, 60 * 1000)
  }

  override def filter(value: Origin): Boolean = {
    if (value != null) {
      value.datetimeOfUTC().toEpochSecond < allowLatest &&
        value.isEdgeLog &&
        value.httpCode() >= 200 && value.httpCode() < 300 &&
        value.httpMethod() != null && Objects.equals(value.httpMethod().toLowerCase(), "get") &&
        DataTimeFilterUtil.domainSecondFilterMin(value, filterDomainsSet, minSecond)
    } else {
      false
    }
  }
}
